// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <melon/executor.h>
#include <memory>

#include <pollux/connectors/connector.h>
#include <pollux/connectors/hive/file_handle.h>
#include <pollux/connectors/hive/hive_config.h>
#include <pollux/connectors/hive/hive_connector_split.h>
#include <pollux/dwio/common/reader.h>

namespace kumo::pollux::connector::hive::iceberg {

struct IcebergDeleteFile;
struct IcebergMetadataColumn;

class PositionalDeleteFileReader {
 public:
  PositionalDeleteFileReader(
      const IcebergDeleteFile& deleteFile,
      const std::string& baseFilePath,
      FileHandleFactory* fileHandleFactory,
      const ConnectorQueryCtx* connectorQueryCtx,
      melon::Executor* executor,
      const std::shared_ptr<const HiveConfig>& hiveConfig,
      const std::shared_ptr<io::IoStatistics>& ioStats,
      const std::shared_ptr<filesystems::File::IoStats>& fsStats,
      dwio::common::RuntimeStatistics& runtimeStats,
      uint64_t splitOffset,
      const std::string& connectorId);

  void readDeletePositions(
      uint64_t baseReadOffset,
      uint64_t size,
      BufferPtr deleteBitmap);

  bool noMoreData();

 private:
  void updateDeleteBitmap(
      VectorPtr deletePositionsVector,
      uint64_t baseReadOffset,
      int64_t rowNumberUpperBound,
      BufferPtr deleteBitmapBuffer);

  bool readFinishedForBatch(int64_t rowNumberUpperBound);

  const IcebergDeleteFile& deleteFile_;
  const std::string& baseFilePath_;
  FileHandleFactory* const fileHandleFactory_;
  melon::Executor* const executor_;
  const ConnectorQueryCtx* connectorQueryCtx_;
  const std::shared_ptr<const HiveConfig> hiveConfig_;
  const std::shared_ptr<io::IoStatistics> ioStats_;
  const std::shared_ptr<filesystems::File::IoStats> fsStats_;
  const std::shared_ptr<filesystems::File::IoStats> fsStats;
  memory::MemoryPool* const pool_;

  std::shared_ptr<IcebergMetadataColumn> filePathColumn_;
  std::shared_ptr<IcebergMetadataColumn> posColumn_;
  uint64_t splitOffset_;

  std::shared_ptr<HiveConnectorSplit> deleteSplit_;
  std::unique_ptr<dwio::common::RowReader> deleteRowReader_;
  // The vector to hold the delete positions read from the positional delete
  // file. These positions are relative to the start of the whole base data
  // file.
  VectorPtr deletePositionsOutput_;
  // The index of deletePositionsOutput_ that indicates up to where the delete
  // positions have been converted into the bitmap
  uint64_t deletePositionsOffset_;
  // Total number of rows read from this positional delete file reader,
  // including the rows filtered out from filters on both filePathColumn_ and
  // posColumn_.
  uint64_t totalNumRowsScanned_;
};

} // namespace kumo::pollux::connector::hive::iceberg
